package io.xxx.xbutler.core.event.listen;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.hy.corecode.idgen.WFGIdGenerator;
import io.xxx.xbutler.constant.Topics;
import io.xxx.xbutler.core.select.receiver.ReceiverSelector;
import io.xxx.xbutler.core.select.receiver.ReceiverSelectorFactory;
import io.xxx.xbutler.data.TaskMapper;
import io.xxx.xbutler.data.TaskReceiverMapper;
import io.xxx.xbutler.domain.SelectMode;
import io.xxx.xbutler.domain.task.Receiver;
import io.xxx.xbutler.domain.task.Task;
import io.xxx.xbutler.domain.task.TaskReceiver;
import io.xxx.xbutler.domain.task.TaskStatus;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.pulsar.annotation.PulsarListener;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.time.LocalDateTime;
import java.util.Objects;
import java.util.Random;
import java.util.concurrent.ThreadLocalRandom;

@Slf4j
@Component
public class TaskExecuteListener {

    private final Random random = ThreadLocalRandom.current();

    @Resource
    private TaskMapper taskMapper;

    @Resource
    private ReceiverSelectorFactory selectorFactory;

    @Resource
    private WFGIdGenerator idGenerator;

    @Resource
    private TaskReceiverMapper taskReceiverMapper;

    @Value("${push.disperse.enabled:true}")
    private Boolean dispersed;

    @PulsarListener(topics = Topics.TASK_EXECUTE)
    public void received(Long taskId) {
        log.info("任务开始执行[{}]", taskId);

        Task task = taskMapper.selectById(taskId);
        task.setStatus(TaskStatus.RUNNING)
                .setUpdatedTime(LocalDateTime.now());
        taskMapper.updateById(task);

        SelectMode selectMode = task.getSelectMode();
        ReceiverSelector selector = selectorFactory.getSelector(selectMode);
        if (selector == null) {
            log.info("ReceiverSelector为空[taskId:{}, selectMode:{}]", taskId, selectMode);
            return;
        }

        int betweenSeconds;
        if (dispersed) {
            betweenSeconds = (int) Duration.between(task.getStartTime(), task.getEndTime()).abs().toSeconds();
        } else {
            betweenSeconds = 0;
        }
        Long pushCount = selector.select(task)
                .doOnNext(it -> {
                    TaskReceiver taskReceiver = buildTaskReceiver(it, task, betweenSeconds);
                    if (!exists(taskReceiver)) {
                        taskReceiverMapper.insert(taskReceiver);
                    }
                })
                .count()
                .block();
        task.setPushCount(Objects.requireNonNull(pushCount).intValue());
        task.setUpdatedTime(LocalDateTime.now());
        if (pushCount == 0) {
            task.setPushSuccessCount(0)
                    .setPushFailCount(0)
                    .setProcess(100.0)
                    .setStatus(TaskStatus.COMPLETED);
        }
        taskMapper.updateById(task);
    }

    private boolean exists(TaskReceiver taskReceiver) {
        LambdaQueryWrapper<TaskReceiver> wrapper = new LambdaQueryWrapper<TaskReceiver>()
                .eq(TaskReceiver::getTaskId, taskReceiver.getTaskId())
                .eq(TaskReceiver::getReceiverType, taskReceiver.getReceiverType())
                .eq(TaskReceiver::getReceiverId, taskReceiver.getReceiverId());
        return taskReceiverMapper.exists(wrapper);
    }

    private TaskReceiver buildTaskReceiver(Receiver receiver, Task task, int betweenSeconds) {
        LocalDateTime now = LocalDateTime.now();
        int offset = 0;
        if (dispersed) {
            offset = random.nextInt(betweenSeconds);
        }
        return new TaskReceiver()
                .setId(idGenerator.next())
                .setTaskId(task.getId())
                .setStoreId(receiver.getStoreId())
                .setStoreCode(receiver.getStoreCode())
                .setStoreName(receiver.getStoreName())
                .setReceiverType(receiver.getType())
                .setReceiverId(receiver.getId())
                .setReceiverSerialNo(receiver.getSerialNo())
                .setReceiverName(receiver.getName())
                .setPushTime(dispersed ? task.getStartTime().plusSeconds(offset) : task.getStartTime())
                .setPushed(false)
                .setCreatedTime(now)
                .setUpdatedTime(now);
    }
}
